package com.cn;

import com.cn.file.MessageStore;
import com.cn.producer.Producer;
import com.cn.producer.SendResult;
import com.cn.producer.impl.ProducerImpl;

import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestProducer {


    public static void main(String[] args) {
        test0();
    }

    public static void test0() {
        Producer defaultProducer = new ProducerImpl("127.0.0.1:2020");
        try {
            defaultProducer.start();
            System.out.println("执行开始");
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(50);
            for (int t = 0; t < 20; t++) {
                executorService.scheduleAtFixedRate(new Runnable() {
                    int i = 0;

                    @Override
                    public void run() {
//                        System.out.println("开始执行了");
                        long t1 = System.currentTimeMillis();
                        String body = i + "->hi,cn_yang[" + Thread.currentThread().getId() + "]";
                        MessageStore messageStore = new MessageStore();
                        messageStore.setTopic("topic").setBody(body.getBytes(Charset.forName("UTF-8")));
                        try {
                            SendResult result = defaultProducer.send(messageStore);
                            if (result != null) {
                                System.out.println(result.getSendStatus() + "==>>" + result.getMsgId());
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
//                        long t2 = System.currentTimeMillis();
//                        System.out.println((t2 - t1) / 1000 + "秒");
                        i++;
                    }
                }, 10, 10, TimeUnit.SECONDS);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void test1() {
        String addr = "127.0.0.1:20200";
        Producer defaultProducer = new ProducerImpl("127.0.0.1:20200");
        try {
            defaultProducer.start();
            System.out.println("执行开始");
//            Thread.sleep(1500);


            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);
            for (int t = 0; t < 1; t++) {
                executorService.schedule(new Runnable() {
                    @Override
                    public void run() {
                        long t1 = System.currentTimeMillis();
                        for (int i = 1; i <= 100000; i++) {
                            String body = i + "->hi,cn_yang[" + Thread.currentThread().getId() + "]";
                            MessageStore messageStore = new MessageStore();
                            messageStore.setTopic("topic").setBody(body.getBytes(Charset.forName("UTF-8")));
                            messageStore.setQueueId(getRandom(0, 3));
                            SendResult result = defaultProducer.send(messageStore);
                            if (result != null) {
                                System.out.println(result.getSendStatus() + "==>>" + result.getMsgId());
                            }
                        }
                        long t2 = System.currentTimeMillis();
                        System.out.println((t2 - t1) / 1000 + "秒");
                    }
                }, 5, TimeUnit.SECONDS);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static int getRandom(int x, int y) {
        return x + (int) (Math.random() * (y - x + 1));
    }


}
